/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2016-2026 hmc,Inc.All Rights Reserved.
 */
package com.sys.midware.MQ.rocketMQ.logic.impl;

import java.util.ArrayList;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListener;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.sys.midware.MQ.rocketMQ.config.ConfigRocketUtil;
import com.sys.midware.MQ.rocketMQ.logic.IPushConsumer;
import com.sys.midware.assist.utils.string.StringUtil;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * <pre>
*   IConsumer consumer = new ConsumerImpl();   
*   consumer.register(messageListener, topics);    
*   或者 
*   consumer.register(listener);
*   consumer.subscribe(topic, tags);
*   注意： 
*   consumer.start();  
*   consumer.shutdown();
 * </pre>
 * 
 * <b>ClassName：</b> ConsumerImpl <br/>
 * <b>Description：</b> rocketmq消费者类，使用示例 <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午3:54:00 <br/>
 * <b>@version: </b> <br/>
 */
public class PushConsumerImpl implements IPushConsumer {

    private static final Logger logger = LoggerFactory.getLogger(PushConsumerImpl.class);

    private final DefaultMQPushConsumer mqConsumer;

    private List<String> topics;

    /**
     * 
     * @param consumerGroup
     */
    public PushConsumerImpl(String consumerGroup) {
        this(null, consumerGroup);
    }

    public PushConsumerImpl(String nameServer, String consumerGroup) {
        this(nameServer, null, consumerGroup);

    }

    private void init(DefaultMQPushConsumer mqConsumer, String nameServer, String clientIp, String instance,
            String consumerGroup) {
        if (nameServer == null) {
            nameServer = ConfigRocketUtil.getRocketNameServer();
        }

        if (instance == null) {
            instance = consumerGroup;
        }
        mqConsumer.setNamesrvAddr(nameServer);
        mqConsumer.setInstanceName(instance);
        // 配置集群消费
        mqConsumer.setMessageModel(MessageModel.CLUSTERING);
        if (StringUtil.isNotEmpty(clientIp)) {
            mqConsumer.setClientIP(clientIp);
        }

        // mqConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

    }

    public PushConsumerImpl(String nameServer, String instance, String consumerGroup) {
        mqConsumer = new DefaultMQPushConsumer(consumerGroup);
        topics = new ArrayList<String>();
        init(mqConsumer, nameServer, null, instance, consumerGroup);

        // if (nameServer == null) {
        // nameServer = ConfigRocketUtil.getRocketNameServer();
        // }
        //
        // if (instance == null) {
        // instance = consumerGroup;
        // }
        // mqConsumer.setNamesrvAddr(nameServer);
        // mqConsumer.setInstanceName(instance);
        // //配置集群消费
        // mqConsumer.setMessageModel(MessageModel.CLUSTERING);
        //// mqConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // topics = new ArrayList<String>();
    }

    public PushConsumerImpl(String nameServer,String instance,String consumerGroup,String clientIp) {
        mqConsumer = new DefaultMQPushConsumer(consumerGroup);
        topics = new ArrayList<String>();
        init(mqConsumer, nameServer, clientIp, instance, consumerGroup);
    }

    public void subscribe(List<String> topics) {
        for (String topic : topics) {
            topic = topic.trim();
            this.topics.add(topic);
            try {
                mqConsumer.subscribe(topic, "*");
            } catch (MQClientException e) {
                logger.error("subscribe topic " + topic.toString() + " on error!", e);
            }
        }
    }

    @Override
    public void subscribe(String topic, String tags) {
        try {
            topic = topic.trim();
            topics.add(topic);
            mqConsumer.subscribe(topic, tags);
        } catch (MQClientException e) {
            logger.error("subscribe topic " + topic.toString() + " on error!", e);
        }
    }

    public DefaultMQPushConsumer getMQConsumer() {
        return this.mqConsumer;
    }

    public void setNameServer(String namesrvAddr) {
        mqConsumer.setNamesrvAddr(namesrvAddr);
    }

    @SuppressWarnings("deprecation")
    @Override
    public void register(MessageListener messageListener) {
        if (messageListener instanceof MessageListenerConcurrently) {// 消息并行的方式
            MessageListenerConcurrently messageListenerConcurrently = (MessageListenerConcurrently) messageListener;
            mqConsumer.registerMessageListener(messageListenerConcurrently);
        } else if (messageListener instanceof MessageListenerOrderly) {// 推荐使用顺序串行的方式保持消息的顺序性
            MessageListenerOrderly messageListenerOrderly = (MessageListenerOrderly) messageListener;
            mqConsumer.registerMessageListener(messageListenerOrderly);
        } else {// 业务尽量不试用该方式
            mqConsumer.registerMessageListener(messageListener);
        }

    }

    @Override
    public void register(MessageListener messageListener, List<String> topics) {
        register(messageListener);
        subscribe(topics);
    }

    @Override
    public void register(MessageListener messageListener, String topic, String tags) {
        register(messageListener);
        subscribe(topic, tags);
    }

    @Override
    public void start() {
        try {
            logger.info(" ============= Consumer 准备启动 ==========");
            mqConsumer.start();
            logger.info(" nameServer " + mqConsumer.getNamesrvAddr() + ", topics " + this.topics + " , group "
                    + mqConsumer.getConsumerGroup() + " ");
            logger.info(" ============= Consumer 启动成功 ==========");
        } catch (MQClientException e) {
            logger.error(" ============= Consumer 启动出错 ==========", e);
        }
    }

    @Override
    public void shutdown() {
        mqConsumer.shutdown();
    }

    /**
     * set method do some config for client
     */
    public void setConsumeThreadMin(int consumeThreadMin) {
        mqConsumer.setConsumeThreadMin(consumeThreadMin);
    }

    public void setConsumeThreadMax(int consumeThreadMax) {
        mqConsumer.setConsumeThreadMax(consumeThreadMax);
    }

    public void setPullThresholdForQueue(int pullThresholdForQueue) {
        mqConsumer.setPullThresholdForQueue(pullThresholdForQueue);
    }

    public void setConsumeMessageBatchMaxSize(int consumeMessageBatchMaxSize) {
        mqConsumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
    }

    public void setPullBatchSize(int pullBatchSize) {
        mqConsumer.setPullBatchSize(pullBatchSize);
    }

    public void setConsumeConcurrentlyMaxSpan(int consumeConcurrentlyMaxSpanSize) {
        mqConsumer.setConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpanSize);
    }

    @Override
    public void initConfig() {
        setConsumeThreadMax(ConfigRocketUtil.getMsgConsumerThreadMaxSize());
        setConsumeThreadMin(ConfigRocketUtil.getMsgConsumerThreadMinSize());
        setConsumeMessageBatchMaxSize(ConfigRocketUtil.getMsgConsumerBatchMaxSize());
    }

}
